package com.edata.bigdata.flink

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.cep.functions.PatternProcessFunction
import org.apache.flink.cep.pattern.conditions.IterativeCondition
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala.DataStream

class PatternBuilder[IN, OUT: TypeInformation] {
  var pattern: Pattern[IN, IN] = _
  var patternStream: PatternStream[IN] = _
  //var processedStream: DataStream[OUT] = _

  def begin(name: String, condition: IterativeCondition[IN]): PatternBuilder[IN, OUT] = {
    pattern = Pattern.begin[IN](name).where(condition)
    this
  }

  def followed(name: String, condition: IterativeCondition[IN]): PatternBuilder[IN, OUT] = {
    pattern = pattern.followedBy(name).where(condition)
    this
  }

  def next(name: String, condition: IterativeCondition[IN]): PatternBuilder[IN, OUT] = {
    pattern = pattern.next(name).where(condition)
    this
  }

  def or(name: String, condition: IterativeCondition[IN]): PatternBuilder[IN, OUT] = {
    pattern = pattern.or(condition)
    this
  }

  def until(condition: IterativeCondition[IN]): PatternBuilder[IN, OUT] = {
    pattern = pattern.until(condition)
    this
  }

  def onOrMore(): PatternBuilder[IN, OUT] = {
    pattern.oneOrMore
    this
  }

  def times(num: Int): PatternBuilder[IN, OUT] = {
    pattern.times(num)
    this
  }

  def optional(): PatternBuilder[IN, OUT] = {
    pattern.optional
    this
  }

  def greedy(): PatternBuilder[IN, OUT] = {
    pattern.greedy
    this
  }

  def build(): PatternBuilder[IN, OUT] = {
    this
  }


  def applyToDataStream(ds: DataStream[IN]): PatternBuilder[IN, OUT] = {
    patternStream = CEP.pattern(ds, pattern).inProcessingTime()
    this
  }

  def applyByProcessFunction(func: PatternProcessFunction[IN, OUT]): DataStream[OUT] = {
    patternStream.process(func)
  }
}
